package com.to8to.kitt.esb.http;

import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_LENGTH;
import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.DefaultFullHttpRequest;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.QueryStringDecoder;

import java.util.Map.Entry;

import org.apache.thrift.TBase;
import org.apache.thrift.protocol.TMessage;
import org.apache.thrift.protocol.TMessageMeta;
import org.apache.thrift.protocol.TMessageType;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.to8to.commons.json.JSONArray;
import com.to8to.commons.json.JSONObject;
import com.to8to.commons.json.JSONValue;
import com.to8to.kitt.ThriftClassFactory;
import com.to8to.kitt.ThriftMessage;
import com.to8to.kitt.esb.BusChannelHandler;
import com.to8to.kitt.esb.ZKServiceChannelPool;
import com.to8to.kitt.utils.LogUtils;
import com.to8to.kitt.utils.NettyUtils;
import com.to8to.kitt.utils.ThriftUtils;
import com.to8to.kitt.utils.UniqueIdGenerator;


@Sharable
public class Http2ThriftHandler extends ChannelHandlerAdapter {
    
	public static Logger logger = LoggerFactory.getLogger(Http2ThriftHandler.class);
	
    ObjectMapper mapper = new ObjectMapper();
    
    ZKServiceChannelPool serviceChannelPool;
    
    public Http2ThriftHandler()
    {
    	
    }
    
    public Http2ThriftHandler(ZKServiceChannelPool serviceChannelPool)
    {
    	this.serviceChannelPool = serviceChannelPool;
    }
    
    
    public ZKServiceChannelPool getServiceChannelPool() {
		return serviceChannelPool;
	}


	public void setServiceChannelPool(ZKServiceChannelPool serviceChannelPool) {
		this.serviceChannelPool = serviceChannelPool;
	}


	public String getString(JsonNode node, String key)
    {
    	JsonNode n = node.findValue(key);
    	
    	if(n == null)
    		return null;
    	
    	return n.asText();
    }
	
	@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		
		long t1 = System.currentTimeMillis();
		
        if (msg instanceof DefaultFullHttpRequest) 
        {
        	DefaultFullHttpRequest req = (DefaultFullHttpRequest) msg;
        	
//        	JsonNode node = null;
        	
        	JSONObject node = null;
        	
        	if(req.getMethod() == HttpMethod.GET)
            {
        		req.getDecoderResult();
        		String uri = req.getUri();
        		QueryStringDecoder decoder = new QueryStringDecoder(uri);
        		String d = decoder.parameters().get("d").get(0);
//        		node = mapper.readTree(d);
        		
        		node = JSONValue.parseJSONObject(d);
            }
        	
            if (req.getMethod() == HttpMethod.POST) 
            {   
            	ByteBuf bytebuf = req.content();
            	
            	byte[] array = new byte[bytebuf.readableBytes()];
            	bytebuf.getBytes(0, array);
            	
//            	node = mapper.readTree(array);
            	
            	node = JSONValue.parseJSONObject(new String(array)); 
            }
            
        	try
        	{  
	        	String ns = node.getString("service");
	        	
	        	String service = HttpBusServer.nsMap.get(ns);
	        	
	        	if(service == null)
	        	{
        			logger.error("命名空间找不到: " + ns);
        			sendResponse(ctx, HttpResponseStatus.BAD_REQUEST, "命名空间找不到: " + ns);
        			return;
	        	}
	        	
	        	String method = node.getString("interface");
	        	
	        	String tag = node.getString("tag");
	        	
	        	LogUtils.openTag(tag);
	        	
        		logger.debug("===> 收到客户端请求! 消息链路:{} 对方地址:{} \n 消息体:{}", ctx.channel(), ctx.channel().remoteAddress(), node.toString());
	        	
	        	String tagSuffix = "";
	        	
	        	if(tag != null)
	        	{
	        		LogUtils.openTag(tag);
	        		tagSuffix = ":" + tag;
	        	}
	        	
	        	String name = service + ":" + method + tagSuffix;
	        	
	        	TMessageMeta meta = new TMessageMeta(name);
	        	
	        	int seqid = UniqueIdGenerator.uniqueInt();
	        	
	        	String msgKey = BusChannelHandler.genKey(name, seqid);
	        	
//	        	logger.error("keytest:生成key:{}", msgKey);
	        	
            	ThriftMessage inmsg = new ThriftMessage();
            	
            	long t2 = System.currentTimeMillis();
            	
            	Class cls = ThriftClassFactory.loadThriftClass(service + "$" + method + "_args");
        		
        		if(cls == null)
        		{
        			logger.error("加载类失败: {}", service + "$" + method + "_args");
        			sendResponse(ctx, HttpResponseStatus.BAD_REQUEST, "加载类失败: " + service + "$" + method + "_args");
        			return;
        		}
            	
        		JSONArray argsNode = node.getJSONArray("args");
        		
        		TBase targs = null;
        		try{
        			targs = ThriftUtils.JsonArray2ThriftArgs(argsNode, cls);
        		}
        		catch(Exception e)
        		{
        			logger.error("参数解析错误:" + node.toString() + "\nException:" + LogUtils.logStackTrace(e));
        			sendResponse(ctx, HttpResponseStatus.BAD_REQUEST, "参数解析错误:" + node.toString() + "\nException:" + LogUtils.logStackTrace(e));
        			return;
        		}
            	
        		long t3 = System.currentTimeMillis();
        		
        		if(t3-t2 > 100)
        		{
        			logger.warn("构造thrift对象耗时比较久: {} ms, 构造类:{}", (t3-t2), cls);
        		}
            	
            	inmsg.setHead(new TMessage(name, TMessageType.CALL, seqid));
                
            	inmsg.setBody(targs);
            	
            	inmsg.setMeta(meta);
            	
            	logger.debug("thrift消息构造成功：{}", inmsg);
            	
            	HttpBusMessageState status = new HttpBusMessageState(msgKey, 0, seqid, meta, ctx, inmsg);
            	
            	status.setRecvFromClientTime(t1);
            	
	        	String host = node.getString("host");
	        	
	        	Entry<String, Channel> entry = null;
	        	
	        	if(host != null)
	        	{
	        		logger.debug("指定服务地址: " + host);
	        		
	        		entry = serviceChannelPool.getChannelMT(service, host);
	        	}
	        	else 
	        		entry = serviceChannelPool.getChannelMT(service);
	        	
	        	if(entry == null)
	    		{
	    			logger.error("路由失败,找不到服务地址: " + service);
	    			
	    			sendResponse(ctx, HttpResponseStatus.SERVICE_UNAVAILABLE, "路由失败,找不到服务地址: " + service);
	    			
	    			return;
	    		}
        	
            	ChannelFutureListener l = new HttpSDFutureListener(entry, status, serviceChannelPool);
            	
            	status.recordStartSendDownTime();
            	
            	HttpBusServer.putState(msgKey, status);
            	
            	ChannelFuture cf = entry.getValue().writeAndFlush(inmsg);
            	
            	cf.addListener(l);
            	
            	LogUtils.removeTag();
        	}
        	catch(Throwable e)
        	{
        		NettyUtils.exitIfOutOfMemory(e);
        		String txt = e.toString();
        		
        		sendResponse(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR, txt);
        	}
        }
    }
	
	public static void sendResponse(ChannelHandlerContext ctx, HttpResponseStatus status, String txt, ChannelFutureListener l)
	{
		if(status != HttpResponseStatus.OK)
			txt = "{\"fail\":\" "+ JSONObject.escape(txt) +"\"}";
    	FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, status, Unpooled.wrappedBuffer(txt.getBytes()));
        response.headers().set(CONTENT_TYPE, "text/plain");
        response.headers().set(CONTENT_LENGTH, response.content().readableBytes());
        ctx.writeAndFlush(response).addListener(l);
	}
	
	public static void sendResponse(ChannelHandlerContext ctx, HttpResponseStatus status, String txt)
	{
		sendResponse(ctx, status, txt, ChannelFutureListener.CLOSE);
	}

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

    	NettyUtils.exitIfOutOfMemory(cause);
    }
}